//! # Compose Module
//!
//! 提供函数式编程风格的组合工具，用于构建和组合异步计算。
//!
//! 该模块的核心是 `CallBox` 类型，它代表一个异步计算，接收一个输入并产生一个输出。
//! `Composer` 结构体提供了各种方法来创建和组合 `CallBox` 实例，使得可以以声明式的方式
//! 构建复杂的异步计算流程。
//!
//! ## 主要功能
//!
//! * 创建异步计算：使用 `Composer::from_fn` 从异步闭包创建 `CallBox`
//! * 组合异步计算：使用 `Composer::compose` 和 `Composer::compose3` 等方法串联多个 `CallBox`
//! * 并行执行：使用 `Composer::compose_parallel_base` 和相关方法并行执行多个 `CallBox`
//! * 错误处理：使用 `Composer::make_retry_callbox` 实现重试逻辑
//! * 集合处理：使用 `ListHandler` 和 `Composer::make_seq_handler_callbox` 处理集合元素
//!
//! ## 示例
//!
//! ```rust
//! use common::compose::{Composer, make_empty_ctx};
//!
//! // 创建一个简单的异步计算
//! let add_one = Composer::from_fn(|x: i32| async move { Ok(x + 1) });
//!
//! // 创建另一个异步计算
//! let multiply_by_two = Composer::from_fn(|x: i32| async move { Ok(x * 2) });
//!
//! // 组合这两个计算
//! let composed = Composer::compose(add_one, multiply_by_two);
//!
//! // 执行组合后的计算
//! let result = composed.run(5, make_empty_ctx()).await.unwrap();
//! assert_eq!(result, 12); // (5 + 1) * 2 = 12
//! ```

use std::{any::Any, sync::Arc};

use futures::Future;
use async_trait::async_trait;


/// 函数式计算中的错误类型
///
/// 该枚举定义了两种错误类型：
/// - `Generic`: 一般错误，不需要特殊处理
/// - `RetryableError`: 可重试的错误，在重试机制中会被特殊处理
#[derive(Debug, thiserror::Error)]
pub enum FError {
    /// 一般错误，包裹了一个标准错误类型
    #[error("{0}")]
    Generic(Box<dyn std::error::Error + Send + Sync>),

    /// 可重试的错误，在重试机制中会被特殊处理
    #[error("Retryable Error: {0}")]
    RetryableError(Box<dyn std::error::Error + Send + Sync>),
}


/// 可重试错误的特征接口
///
/// 实现该特征的错误类型可以自定义判断错误是否可重试的逻辑
/// 这在需要特定错误处理策略的场景下非常有用
pub trait RetryableError: std::error::Error + Send + Sync + 'static {
    /// 判断给定的错误是否可重试
    fn is_retryable(error: &(dyn std::error::Error + 'static)) -> bool;
}

/// 调用上下文接口
///
/// 该接口定义了一个上下文对象，可以在异步计算中传递额外的信息
pub trait CallContext: Send + Sync {
    /// 根据键获取上下文中的值
    fn get(&self, key: String) -> Result<Box<dyn Any>, FError>;
}

/// 调用上下文的引用类型
pub type ContextBox = Arc<Box<dyn CallContext>>;

/// 空的调用上下文实现
///
/// 这个结构体实现了一个空的上下文，对任何键的访问都会返回错误
struct EmptyCallContext;
impl CallContext for EmptyCallContext {
    fn get(&self, _: String) -> Result<Box<dyn Any>, FError> {
        Err(FError::Generic(Box::new(std::io::Error::new(std::io::ErrorKind::NotFound, "key not found"))))
    }
}

/// 创建一个空的调用上下文
///
/// 当不需要传递额外信息时，可以使用这个函数创建一个空的上下文
pub fn make_empty_ctx() -> ContextBox {
    Arc::new(Box::new(EmptyCallContext))
}

/// 上下文相关的工具类
struct ContextBoxUtils;
impl ContextBoxUtils {
    /// 从函数创建一个调用上下文
    ///
    /// 该方法允许使用自定义函数来实现上下文的值获取逻辑
    pub fn make_from_fn<F>(f: F) -> impl CallContext
    where
        F: Fn(String) -> Result<Box<dyn Any>, FError> + Send + Sync
    {
        struct InnerCallContext<F> {
            f: F,
        }
        impl<F> CallContext for InnerCallContext<F>
        where
            F: Fn(String) -> Result<Box<dyn Any>, FError> + Send + Sync
        {
            fn get(&self, key: String) -> Result<Box<dyn Any>, FError> {
                (self.f)(key)
            }
        }

        InnerCallContext { f }
    }
}


/// 异步计算的核心特征
///
/// 该特征定义了一个异步计算，它接收一个输入类型 `T` 和一个上下文，
/// 并产生一个输出类型 `R` 或错误。这是整个组合系统的基础。
#[async_trait]
pub trait CallCore<T: Send, R: Send> : Send + Sync {
    /// 执行异步计算
    async fn run(&self, arg: T, ctx: ContextBox) -> Result<R, FError>;
}

/// 同步计算的核心特征
///
/// 该特征定义了一个同步计算，它接收一个输入类型 `T`，
/// 并产生一个输出类型 `R` 或错误。
#[async_trait]
pub trait FnCore<T: Send + Sync, R: Send + Sync> : Send + Sync {
    /// 执行同步计算
    fn run(&self, arg: T) -> Result<R, FError>;
}

/// 异步计算的引用类型
///
/// 这是对 `CallCore` 特征对象的引用包装，使其可以在不同的上下文中传递和组合。
pub type CallBox<'a, T, R> = Arc<Box<dyn CallCore<T, R> + 'a>>;

/// 同步计算的引用类型
///
/// 这是对 `FnCore` 特征对象的引用包装。
pub type FnBox<'a, T, R> = Box<dyn FnCore<T, R> + 'a>;

/// 并行处理集合元素的处理器
///
/// 该结构体封装了对集合中每个元素进行并行处理的逻辑，
/// 可以控制并行度。
pub struct ListHandler<'a, T: Send + Sync + 'a, R: Send + Sync + 'a> {
    /// 并行处理的最大数量
    parallel_num: usize,
    /// 处理单个元素的 CallBox
    item_handler: CallBox<'a, T, R>,
}

impl<'a, T: Send + Sync +'a, R: Send + Sync +'a> ListHandler<'a, T, R> {
    /// 创建一个新的 ListHandler
    ///
    /// # 参数
    ///
    /// * `parallel_num` - 并行处理的最大数量
    /// * `item_handler` - 处理单个元素的 CallBox
    ///
    /// # 返回值
    ///
    /// 返回一个 CallBox，它接收一个元素列表，并返回处理结果的列表
    pub fn new(parallel_num: usize, item_handler: CallBox<'a, T, R>) -> CallBox<'a, Vec<T>, Vec<R>> {
        Arc::new(Box::new(Self {
            parallel_num,
            item_handler,
        }))
    }
}
#[async_trait]
impl <'a, T: Send + Sync +'a, R: Send + Sync +'a> CallCore<Vec<T>, Vec<R>> for ListHandler<'a, T, R> {
    /// 并行处理集合中的元素
    ///
    /// 该方法将对输入列表中的每个元素应用 `item_handler`，
    /// 并将结果收集到一个新的列表中。处理过程中的并行度
    /// 由 `parallel_num` 控制。
    async fn run(&self, arg: Vec<T>, ctx: ContextBox) -> Result<Vec<R>, FError> {
        use futures::stream::{self, StreamExt};

        // 使用 StreamExt::buffer_unordered 来限制并行数量
        let results = stream::iter(arg)
            .map(|item| {
                let handler = &self.item_handler;
                let ctx = ctx.clone();
                async move { handler.run(item, ctx).await }
            })
            .buffered(self.parallel_num) // 使用 parallel_num 控制并行度
            .collect::<Vec<Result<R, FError>>>()
            .await;

        // 处理结果
        let mut output = Vec::with_capacity(results.len());
        for result in results {
            match result {
                Ok(r) => output.push(r),
                Err(e) => return Err(e),
            }
        }

        Ok(output)
    }
}

/// 函数式计算的组合器
///
/// `Composer` 提供了一系列静态方法，用于创建和组合 `CallBox` 实例。
/// 这些方法允许以声明式的方式构建复杂的异步计算流程。
pub struct Composer {
}

impl Composer {
    /// 从异步闭包创建 CallBox
    ///
    /// 该方法将一个异步闭包转换为 CallBox，使其可以与其他 CallBox 组合。
    /// 这是创建基本计算单元的最常用方法。
    ///
    /// # 参数
    ///
    /// * `f` - 一个异步闭包，接收类型 T 的输入，返回 Result<R, FError> 的 Future
    ///
    /// # 返回值
    ///
    /// 返回一个 CallBox，它封装了给定的异步闭包
    ///
    /// # 示例
    ///
    /// ```
    /// let add_one = Composer::from_fn(|x: i32| async move { Ok(x + 1) });
    /// ```
    pub fn from_fn<'a, T: Send + Sync + 'a, R: Send + Sync + 'a, F, Fut>(f: F) -> CallBox<'a, T, R>
    where
        Fut: Future<Output = Result<R, FError>> + Send + 'a,
        F: Fn(T) -> Fut + Send + Sync + 'a,
    {
        Self::from_fn2(move |arg, _ctx| f(arg))
    }

    /// 从带上下文的异步闭包创建 CallBox
    ///
    /// 该方法类似于 `from_fn`，但允许闭包访问上下文对象。
    /// 这在需要从上下文中获取额外信息的场景下非常有用。
    ///
    /// # 参数
    ///
    /// * `f` - 一个异步闭包，接收类型 T 的输入和上下文对象，返回 Result<R, FError> 的 Future
    ///
    /// # 返回值
    ///
    /// 返回一个 CallBox，它封装了给定的异步闭包
    pub fn from_fn2<'a, T: Send + Sync + 'a, R: Send + Sync + 'a, F, Fut>(f: F) -> CallBox<'a, T, R>
    where
        Fut: Future<Output = Result<R, FError>> + Send + 'a,
        F: Fn(T, ContextBox) -> Fut + Send + Sync + 'a,
    {
        struct FnRunable1<T, R, F> {
            f: F,
            _phantom: std::marker::PhantomData<(T, R)>,
        }

        #[async_trait]
        impl<'a, T: Send + Sync, R: Send + Sync, F, Fut> CallCore<T, R> for FnRunable1<T, R, F>
        where
            Fut: Future<Output = Result<R, FError>> + Send + 'a,
            F: Fn(T, ContextBox) -> Fut + Send + Sync,
        {
            async fn run(&self, arg: T, ctx: ContextBox) -> Result<R, FError> {
                (self.f)(arg, ctx).await
            }
        }

        Arc::new(Box::new(FnRunable1 {
            f,
            _phantom: std::marker::PhantomData,
        }))
    }

    /// 创建一个恒等 CallBox
    ///
    /// 返回一个 CallBox，它简单地返回输入参数而不做任何修改。
    /// 这在需要一个“不做任何事”的计算单元时非常有用。
    ///
    /// # 返回值
    ///
    /// 返回一个 CallBox，它简单地返回输入参数
    pub fn identity<'a, T: Send + Sync +'a>() -> CallBox<'a, T, T> {
        Self::from_fn(|arg: T| Box::pin(async move { Ok(arg) }))
    }

    /// 使用上下文参数增强 CallBox
    ///
    /// 该方法将输入参数添加到上下文中，使其可以在后续计算中被访问。
    /// 这在需要在计算链中传递额外信息时非常有用。
    ///
    /// # 参数
    ///
    /// * `key` - 在上下文中存储参数的键
    /// * `call` - 要增强的 CallBox
    ///
    /// # 返回值
    ///
    /// 返回一个新的 CallBox，它将输入参数添加到上下文中，然后调用原始 CallBox
    pub fn with_context_params<'a, T, R>(key: String, call: CallBox<'a, T, R>) -> CallBox<'a, T, R>
    where
        T: Send + Sync + Clone +'static,
        R: Send + Sync +'a
    {
        let call_arc = Arc::new(call);

        Self::from_fn2(move |arg: T, ctx: ContextBox| {
            let arg_clone = arg.clone();
            let key_clone = key.clone();
            let ctx = ContextBoxUtils::make_from_fn(move |key_arg: String| {
                if key_clone == key_arg {
                    Ok(Box::new(arg_clone.clone()))
                } else {
                    ctx.get(key_arg)
                }
            });

            let call = call_arc.clone();
            Box::pin(async move {
                // 解引用 Arc 获取内部的 CallBox
                call.run(arg, Arc::new(Box::new(ctx))).await
            })
        })
    }

    /// 从上下文中获取值的 CallBox
    ///
    /// 该方法创建一个 CallBox，它从上下文中获取指定键的值。
    /// 这在需要从上下文中获取之前存储的值时非常有用。
    ///
    /// # 参数
    ///
    /// * `key` - 要获取的值的键
    ///
    /// # 返回值
    ///
    /// 返回一个 CallBox，它从上下文中获取指定键的值
    pub fn with_callbox<T: Send + Sync + Clone + 'static>(key: &str) -> CallBox<'static, (), T> {
        let key_str = key.to_string();

        // 注意：这个实现使用 todo!()，可能需要完成
        struct WithHandler<T: Clone + Send + Sync + 'static> {
            key: String,
            _phantom: std::marker::PhantomData<T>,
        }

        #[async_trait]
        impl<T: Clone + Send + Sync + 'static> CallCore<(), T> for WithHandler<T> {
            async fn run(&self, _: (), _ctx: ContextBox) -> Result<T, FError> {
                todo!()
            }
        }

        Arc::new(Box::new(WithHandler {
            key: key_str,
            _phantom: std::marker::PhantomData,
        }))
    }


    /// 组合两个 CallBox
    ///
    /// 该方法将两个 CallBox 组合成一个新的 CallBox，其中第一个 CallBox 的输出
    /// 作为第二个 CallBox 的输入。这是函数式编程中的基本组合操作。
    ///
    /// # 参数
    ///
    /// * `f1` - 第一个 CallBox，接收类型 T 的输入，返回类型 T2 的输出
    /// * `f2` - 第二个 CallBox，接收类型 T2 的输入，返回类型 R 的输出
    ///
    /// # 返回值
    ///
    /// 返回一个新的 CallBox，它接收类型 T 的输入，返回类型 R 的输出
    ///
    /// # 示例
    ///
    /// ```
    /// let add_one = Composer::from_fn(|x: i32| async move { Ok(x + 1) });
    /// let multiply_by_two = Composer::from_fn(|x: i32| async move { Ok(x * 2) });
    /// let composed = Composer::compose(add_one, multiply_by_two);
    /// // composed 将先对输入加 1，然后将结果乘以 2
    /// ```
    pub fn compose<'a, T, R, T2> (f1: CallBox<'a, T, T2>, f2: CallBox<'a, T2, R>) -> CallBox<'a, T, R>
    where
        T: Send + Sync + 'a,
        R: Send + Sync + 'a,
        T2: Send + Sync + 'a,
    {
        struct ComposedRunable<'a, T, R, T2> {
            f1: CallBox<'a, T, T2>,
            f2: CallBox<'a, T2, R>,
        }

        #[async_trait]
        impl<'a, T, R, T2> CallCore<T, R> for ComposedRunable<'a, T, R, T2>
        where
            T: Send + Sync,
            R: Send + Sync,
            T2: Send + Sync
        {
            async fn run(&self, t: T, ctx: ContextBox) -> Result<R, FError> {
                let intermediate = self.f1.run(t, ctx.clone()).await?;
                self.f2.run(intermediate, ctx).await
            }
        }

        Arc::new(Box::new(ComposedRunable { f1, f2 }))
    }

    /// 组合三个 CallBox
    ///
    /// 该方法将三个 CallBox 组合成一个新的 CallBox，其中第一个 CallBox 的输出
    /// 作为第二个 CallBox 的输入，第二个 CallBox 的输出作为第三个 CallBox 的输入。
    /// 这是对 `compose` 方法的便捷扩展。
    ///
    /// # 参数
    ///
    /// * `f1` - 第一个 CallBox，接收类型 T 的输入，返回类型 T1 的输出
    /// * `f2` - 第二个 CallBox，接收类型 T1 的输入，返回类型 T2 的输出
    /// * `f3` - 第三个 CallBox，接收类型 T2 的输入，返回类型 R 的输出
    ///
    /// # 返回值
    ///
    /// 返回一个新的 CallBox，它接收类型 T 的输入，返回类型 R 的输出
    ///
    /// # 示例
    ///
    /// ```
    /// let add_one = Composer::from_fn(|x: i32| async move { Ok(x + 1) });
    /// let multiply_by_two = Composer::from_fn(|x: i32| async move { Ok(x * 2) });
    /// let subtract_three = Composer::from_fn(|x: i32| async move { Ok(x - 3) });
    /// let composed = Composer::compose3(add_one, multiply_by_two, subtract_three);
    /// // composed 将先对输入加 1，然后将结果乘以 2，最后减去 3
    /// ```
    pub fn compose3<'a, T, R, T1, T2> (
        f1: CallBox<'a, T, T1>,
        f2: CallBox<'a, T1, T2>,
        f3: CallBox<'a, T2, R>
    ) -> CallBox<'a, T, R>
    where
        T: Send + Sync + 'static,
        R: Send + Sync + 'static,
        T1: Send + Sync + 'static,
        T2: Send + Sync + 'static
    {
        let composed = Self::compose(f1, f2);
        Self::compose(composed, f3)
    }

    /// 创建一个执行 CallBox 的 CallBox
    ///
    /// 该方法创建一个 CallBox，它接收一个 CallBox 和一个参数，
    /// 然后执行该 CallBox 并返回结果。这在需要动态选择要执行的 CallBox 时非常有用。
    ///
    /// # 返回值
    ///
    /// 返回一个 CallBox，它接收一个 (CallBox, 参数) 元组，执行该 CallBox 并返回结果
    pub fn make_exec<T: Send + Sync + 'static, R: Send + Sync + 'static>() -> CallBox<'static, (CallBox<'static, T, R>, T), R> {
        struct FnA {}
        #[async_trait]
        impl<T: Send + Sync + 'static, R: Send + Sync + 'static> CallCore<(CallBox<'static, T, R>, T), R> for FnA  {
            async fn run(&self, (f, arg): (CallBox<'static, T, R>, T), ctx: ContextBox) -> Result<R, FError> {
                f.run(arg, ctx).await
            }
        }
        Arc::new(Box::new(FnA {}))
    }

    /// 创建一个带重试功能的 CallBox
    ///
    /// 该方法将一个 CallBox 包装成一个新的 CallBox，它在遇到可重试错误时
    /// 会自动重试。这在处理不稳定的外部服务或网络请求时非常有用。
    ///
    /// # 参数
    ///
    /// * `callbox` - 要包装的 CallBox
    /// * `retry_num` - 最大重试次数
    /// * `retry_delay` - 重试间隔时间
    ///
    /// # 返回值
    ///
    /// 返回一个新的 CallBox，它在遇到可重试错误时会自动重试
    ///
    /// # 示例
    ///
    /// ```
    /// let unstable_fn = Composer::from_fn(|_: ()| async {
    ///     // 模拟不稳定的服务
    ///     if rand::random::<f32>() < 0.5 {
    ///         Err(FError::RetryableError(Box::new(std::io::Error::new(
    ///             std::io::ErrorKind::ConnectionReset,
    ///             "connection reset"
    ///         ))))
    ///     } else {
    ///         Ok("success")
    ///     }
    /// });
    ///
    /// let retry_fn = Composer::make_retry_callbox(
    ///     unstable_fn,
    ///     3,  // 最多重试 3 次
    ///     std::time::Duration::from_millis(100)  // 每次重试间隔 100 毫秒
    /// );
    /// ```
    pub fn make_retry_callbox<'a, T, R>(
        callbox: CallBox<'a, T, R>,
        retry_num: usize,
        retry_delay: std::time::Duration,
    ) -> CallBox<'a, T, R>
    where
        T: Send + Sync + Clone + 'a,
        R: Send + Sync + 'a,
    {
        struct RetryRunable<'a, T, R> {
            inner: CallBox<'a, T, R>,
            retry_num: usize,
            retry_delay: std::time::Duration,
        }

        #[async_trait]
        impl<'a, T, R> CallCore<T, R> for RetryRunable<'a, T, R>
        where
            T: Send + Sync + Clone,
            R: Send + Sync,
        {
            async fn run(&self, arg: T, ctx: ContextBox) -> Result<R, FError> {
                let mut attempts = 0;
                loop {
                    attempts += 1;
                    match self.inner.run(arg.clone(), ctx.clone()).await {
                        Ok(result) => return Ok(result),
                        Err(error) => {
                            if attempts > self.retry_num {
                                return Err(FError::Generic(Box::new(error)));
                            }

                            match error {
                                FError::RetryableError(_) => {
                                    if !self.retry_delay.is_zero() {
                                        tokio::time::sleep(self.retry_delay).await;
                                    }
                                    continue;
                                }
                                _ => return Err(error),
                            }
                        }
                    }
                }
            }
        }

        Arc::new(Box::new(RetryRunable {
            inner: callbox,
            retry_num,
            retry_delay,
        }))
    }



    /// 并行组合两个 CallBox 的基础方法
    ///
    /// 该方法将两个 CallBox 并行组合成一个新的 CallBox。与串行组合不同，
    /// 并行组合会同时执行两个 CallBox，并等待它们都完成。这在需要
    /// 并行处理多个独立任务时非常有用。
    ///
    /// # 参数
    ///
    /// * `f1` - 第一个 CallBox，接收类型 T1 的输入，返回类型 R1 的输出
    /// * `f2` - 第二个 CallBox，接收类型 T2 的输入，返回类型 R2 的输出
    ///
    /// # 返回值
    ///
    /// 返回一个新的 CallBox，它接收一个 (T1, T2) 元组，并行执行两个 CallBox，
    /// 然后返回一个 (R1, R2) 元组
    ///
    /// # 示例
    ///
    /// ```
    /// let fetch_user = Composer::from_fn(|user_id: i32| async move {
    ///     // 从数据库获取用户信息
    ///     Ok(format!("User {}", user_id))
    /// });
    ///
    /// let fetch_posts = Composer::from_fn(|user_id: i32| async move {
    ///     // 从数据库获取用户的文章
    ///     Ok(vec![format!("Post 1 by user {}", user_id), format!("Post 2 by user {}", user_id)])
    /// });
    ///
    /// let parallel_fetch = Composer::compose_parallel_base(fetch_user, fetch_posts);
    /// // parallel_fetch 将并行获取用户信息和用户的文章
    /// ```
    pub fn compose_parallel_base<'a, T1, T2, R1, R2>(
        f1: CallBox<'a, T1, R1>,
        f2: CallBox<'a, T2, R2>,
    ) -> CallBox<'a, (T1, T2), (R1, R2)>
    where
        T1: Send + Sync + 'a,
        T2: Send + Sync + 'a,
        R1: Send + Sync + 'a,
        R2: Send + Sync + 'a,
    {
        struct ParallelRunable<'a, T1, T2, R1, R2> {
            f1: CallBox<'a, T1, R1>,
            f2: CallBox<'a, T2, R2>,
        }

        #[async_trait]
        impl<'a, T1: Send + Sync, T2: Send + Sync, R1: Send + Sync, R2: Send + Sync> CallCore<(T1, T2), (R1, R2)>
            for ParallelRunable<'a, T1, T2, R1, R2>
        {
            async fn run(&self, t: (T1, T2), ctx: ContextBox) -> Result<(R1, R2), FError> {
                let (t1, t2) = t;

                let (r1, r2) = tokio::join!(
                    self.f1.run(t1, ctx.clone()),
                    self.f2.run(t2, ctx)
                );

                match (r1, r2) {
                    (Ok(r1), Ok(r2)) => Ok((r1, r2)),
                    (Err(e), _) => Err(e),
                    (_, Err(e)) => Err(e),
                }
            }
        }

        Arc::new(Box::new(ParallelRunable { f1, f2 }))
    }

    pub fn compose_parallel<'a, T, R1, R2>(
        f1: CallBox<'a, T, R1>,
        f2: CallBox<'a, T, R2>,
    ) -> CallBox<'a, T, (R1, R2)>
    where
        T: Send + Sync + Clone + 'a,
        R1: Send + Sync + 'a,
        R2: Send + Sync + 'a,
    {
        let f_base = Self::compose_parallel_base(f1, f2);
        struct ParallelRunable<'a, T, R1, R2> {
            f_base: CallBox<'a, (T, T), (R1, R2)>,
        }

        #[async_trait]
        impl<'a, T: Send + Sync + Clone, R1: Send + Sync, R2: Send + Sync> CallCore<T, (R1, R2)>
            for ParallelRunable<'a, T, R1, R2>
        {
            async fn run(&self, t: T, ctx: ContextBox) -> Result<(R1, R2), FError> {
                let t_copy = t.clone();
                let (r1, r2) = self.f_base.run((t_copy, t), ctx).await?;
                Ok((r1, r2))
            }
        }

        Arc::new(Box::new(ParallelRunable { f_base }))
    }

    pub fn compose_parallel_simple<'a, T, R1, R2>(
        f1: CallBox<'a, (), R1>,
        f2: CallBox<'a, T, R2>,
    ) -> CallBox<'a, T, (R1, R2)>
    where
        T: Send + Sync  + 'a,
        R1: Send + Sync + 'a,
        R2: Send + Sync + 'a,
    {
        let f_base = Self::compose_parallel_base(f1, f2);
        struct ParallelRunable<'a, T, R1, R2> {
            f_base: CallBox<'a, ((), T), (R1, R2)>,
        }

        #[async_trait]
        impl<'a, T: Send + Sync, R1: Send + Sync, R2: Send + Sync> CallCore<T, (R1, R2)>
            for ParallelRunable<'a, T, R1, R2>
        {
            async fn run(&self, t: T, ctx: ContextBox) -> Result<(R1, R2), FError> {
                let (r1, r2) = self.f_base.run(((), t), ctx).await?;
                Ok((r1, r2))
            }
        }

        Arc::new(Box::new(ParallelRunable { f_base }))
    }


    pub fn make_recursive<'a, T, R>(fable: CallBox<'a, (CallBox<'a, T, R>, T), R>) -> CallBox<'a, T, R>
    where
        T: Send + Sync + 'a,
        R: Send + Sync + 'a,
    {
        struct FnA<'a, T, R> {
            fable: Arc<CallBox<'a, (CallBox<'a, T, R>, T), R>>,
        }

        #[async_trait]
        impl<'a, T: Send + Sync + 'a, R: Send + Sync + 'a> CallCore<T, R> for FnA<'a, T, R> {
            async fn run(&self, arg: T, ctx: ContextBox) -> Result<R, FError> {
                // 创建一个自引用的函数对象
                let self_ref: CallBox<T, R> = Arc::new(Box::new(Self {
                    fable: self.fable.clone(),
                }));

                // 调用 fable 并传入自身引用和参数
                self.fable.run((self_ref, arg), ctx).await
            }
        }

        Arc::new(Box::new(FnA {
            fable: Arc::new(fable),
        }))
    }

    pub fn wrap_some<'a, T: Send + Sync + 'a, R: Send + Sync + 'a> (
        f: CallBox<'a, T, R>,
    ) -> CallBox<'a, T, Option<R>> {
        struct SomeRunable<'a, T, R> {
            f: Arc<CallBox<'a, T, R>>,
        }

        #[async_trait]
        impl<'a, T: Send + Sync, R: Send + Sync> CallCore<T, Option<R>> for SomeRunable<'a, T, R> {
            async fn run(&self, arg: T, ctx: ContextBox) -> Result<Option<R>, FError> {
                let rs = self.f.run(arg, ctx).await?;
                Ok(Some(rs))
            }
        }

        Arc::new(Box::new(SomeRunable {
            f: Arc::new(f),
        }))
    }

    /// 创建一个串行处理集合元素的 CallBox
    ///
    /// 该方法创建一个 CallBox，它串行处理集合中的每个元素。与 ListHandler 不同，
    /// 这个方法保证元素按照它们在集合中的顺序一个接一个地处理，而不是并行处理。
    /// 这在需要保证处理顺序或者处理过程中有依赖关系时非常有用。
    ///
    /// # 参数
    ///
    /// * `item_handler` - 处理单个元素的 CallBox
    ///
    /// # 返回值
    ///
    /// 返回一个 CallBox，它串行处理集合中的每个元素，并返回处理结果的列表
    ///
    /// # 示例
    ///
    /// ```
    /// let process_item = Composer::from_fn(|item: i32| async move {
    ///     // 处理单个元素
    ///     Ok(item * 2)
    /// });
    ///
    /// let process_items = Composer::make_seq_handler_callbox(process_item);
    /// // process_items 将串行处理列表中的每个元素
    /// ```
    pub fn make_seq_handler_callbox<'a, T: Send + Sync + 'a, R: Send + Sync + 'a>(
        item_handler: CallBox<'a, T, R>
    ) -> CallBox<'a, Vec<T>, Vec<R>> {
        struct SeqHandler<'a, T: Send + Sync + 'a, R: Send + Sync + 'a> {
            item_handler: CallBox<'a, T, R>,
        }

        #[async_trait]
        impl<'a, T: Send + Sync + 'a, R: Send + Sync + 'a> CallCore<Vec<T>, Vec<R>> for SeqHandler<'a, T, R> {
            async fn run(&self, items: Vec<T>, ctx: ContextBox) -> Result<Vec<R>, FError> {
                let mut results = Vec::with_capacity(items.len());

                for item in items {
                    let result = self.item_handler.run(item, ctx.clone()).await?;
                    results.push(result);
                }

                Ok(results)
            }
        }

        Arc::new(Box::new(SeqHandler { item_handler }))
    }

    /// 包装可选值的 CallBox
    ///
    /// 该方法创建一个 CallBox，它可以处理 Option<T> 类型的输入。当输入是 Some(T) 时，
    /// 它会使用 fsome 处理值；当输入是 None 时，它会使用 fnone 处理。
    /// 这在需要处理可能不存在的值时非常有用。
    ///
    /// # 参数
    ///
    /// * `fsome` - 处理 Some(T) 值的 CallBox
    /// * `fnone` - 处理 None 值的 CallBox
    ///
    /// # 返回值
    ///
    /// 返回一个新的 CallBox，它可以处理 Option<T> 类型的输入
    ///
    /// # 示例
    ///
    /// ```
    /// let process_value = Composer::from_fn(|value: i32| async move {
    ///     Ok(format!("Value: {}", value))
    /// });
    ///
    /// let handle_none = Composer::from_fn(|_: ()| async move {
    ///     Ok("No value".to_string())
    /// });
    ///
    /// let process_option = Composer::wrap_option(process_value, handle_none);
    /// // process_option 可以处理 Option<i32> 类型的输入
    /// ```
    pub fn wrap_option<'a, T: Send + Sync + 'a, R: Send + Sync + 'a> (
        fsome: CallBox<'a, T, R>,
        fnone: CallBox<'a, (), R>
    ) -> CallBox<'a, Option<T>, R> {
        struct OptionRunable<'a, T, R> {
            fsome: CallBox<'a, T, R>,
            fnone: CallBox<'a, (), R>
        }

        #[async_trait]
        impl<'a, T: Send + Sync, R: Send + Sync> CallCore<Option<T>, R> for OptionRunable<'a, T, R> {
            async fn run(&self, t: Option<T>, ctx: ContextBox) -> Result<R, FError> {
                match t {
                    Some(value) => self.fsome.run(value, ctx).await,
                    None => self.fnone.run((), ctx).await,
                }
            }
        }

        Arc::new(Box::new(OptionRunable { fsome, fnone }))
    }
}






#[cfg(test)]
mod tests {
    use log::info;

    use super::*;
    use std::time::Instant;

    #[tokio::test]
    async fn test_parallel_base_execution() {
        env_logger::init();
        // 创建一个会阻塞几秒钟的函数
        let slow_fn = Composer::from_fn(|_: ()| {
            Box::pin(async {
                info!("开始执行慢函数...");
                tokio::time::sleep(tokio::time::Duration::from_secs(2)).await;
                info!("慢函数执行完成");
                Ok("慢函数结果")
            })
        });

        // 创建一个只打印日志的快速函数
        let fast_fn = Composer::from_fn(|_: ()| {
            Box::pin(async {
                info!("开始执行快函数...");
                info!("快函数执行完成");
                Ok("快函数结果")
            })
        });

        // 使用 compose_parallel_base 组合这两个函数
        let parallel_fn = Composer::compose_parallel_base(slow_fn, fast_fn);

        // 记录开始时间
        let start = Instant::now();

        // 执行并行函数
        let ctx = make_empty_ctx();
        let result = parallel_fn.run(((), ()), ctx).await;

        // 验证结果
        assert!(result.is_ok());
        let (slow_result, fast_result) = result.unwrap();
        assert_eq!(slow_result, "慢函数结果");
        assert_eq!(fast_result, "快函数结果");

        // 验证总执行时间接近慢函数的执行时间
        let elapsed = start.elapsed();
        println!("总执行时间: {:?}", elapsed);
        assert!(elapsed.as_secs() >= 2 && elapsed.as_secs() < 3, "执行时间应该接近2秒");
    }


    #[tokio::test]
    async fn test_from_fn() {
        async fn fn_a(arg: String) -> Result<String, FError> {
            Ok(format!("{}: A", arg))
        }
        let composed = Composer::from_fn(fn_a);

        let composed = Composer::compose(
            Composer::compose(
                composed,
                Composer::from_fn(async |s| Ok(((1, 3), (s, "88"), (3.6), (ListHandler::new(3, Composer::from_fn(fn_a))))))
            ),
            Composer::from_fn(async |args| {
                let (_, (s, _), _, _) = args;
                Ok(format!("{}", s))
            })
        );


        let result = composed.run("test".to_string(), make_empty_ctx()).await;
        assert_eq!(result.unwrap(), "test: A".to_string());
    }


    #[tokio::test]
    async fn test_option() {
        struct FnA {}
        struct FnB {}
        #[async_trait]
        impl CallCore<String, String> for FnA  {
            async fn run(&self, arg: String, _: ContextBox) -> Result<String, FError> {
                Ok(format!("{}: A", arg))
            }

        }

        #[async_trait]
        impl CallCore<(), String> for FnB  {
            async fn run(&self, _: (), _: ContextBox) -> Result<String, FError> {
                Ok("None".to_string())
            }
        }
        let composed = Composer::wrap_option(Arc::new(Box::new(FnA {})), Arc::new(Box::new(FnB {})));
        let result = composed.run(Some("test".to_string()), make_empty_ctx()).await;
        assert_eq!(result.unwrap(), "test: A".to_string());
    }

    #[tokio::test]
    async fn test_ss() {

        struct AStruct {}
        #[async_trait]
        impl CallCore<(CallBox<'static, i32, i32>, i32), i32> for AStruct {
            async fn run(&self, (f, x): (CallBox<'static, i32, i32>, i32), ctx: ContextBox) -> Result<i32, FError> {
                let rs = match x {
                    0 => 1,
                    _ => x * f.run(x - 1, ctx).await?,
                };
                Ok(rs)
            }
        }
        let factorial = Composer::make_recursive(Arc::new(Box::new(AStruct{})));

        let result = factorial.run(5, make_empty_ctx()).await.unwrap();
        assert_eq!(result, 120);
        // let fib = r(boxed_fib);
    }

    #[test]
    fn test_y_combine() {
        fn r<A, R>(g: impl Fn(&dyn Fn(A) -> R, A) -> R) -> impl Fn(A) -> R {
            fn r_inner<'a, A, R>(g: &'a dyn Fn(&dyn Fn(A) -> R, A) -> R) -> impl Fn(A) -> R + 'a {
                move |x| g(&r_inner(g), x)
            }
            move |x| r_inner(&g)(x)
        }
        let g = |f: &dyn Fn(usize) -> usize, x| match x {
            0 => 1,
            _ => x * f(x - 1),
        };

        let fact = r(g);
        println!("{}", fact(5)); // 将会输出 120
    }

    #[tokio::test]
    async fn test_composer() {
        struct FnA {}
        struct FnB {}

        #[async_trait]
        impl CallCore<String, String> for FnA  {
            async fn run(&self, arg: String, _: ContextBox) -> Result<String, FError> {
                Ok(format!("{}: A", arg))
            }
        }

        #[async_trait]
        impl CallCore<String, String> for FnB  {
            async fn run(&self, arg: String, _: ContextBox) -> Result<String, FError> {
                Ok(format!("{}: B", arg))
            }
        }
        let composed = Composer::compose(Arc::new(Box::new(FnA {})), Arc::new(Box::new(FnB {})));
        let result = composed.run("test".to_string(), make_empty_ctx()).await;
        assert_eq!(result.unwrap(), "test: A: B".to_string());
    }


    #[tokio::test]
    async fn test_retry_callbox() {

        #[derive(Debug)]
        struct TestError(String);

        impl std::fmt::Display for TestError {
            fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
                write!(f, "TestError: {}", self.0)
            }
        }

        impl std::error::Error for TestError {}

        let counter = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let counter_clone = counter.clone();

        // 添加明确的返回类型标注
        let failing_fn: CallBox<'static, (), &'static str> = Composer::from_fn(move |_: ()| {
            let counter = counter_clone.clone();
            Box::pin(async move {
                let attempts = counter.fetch_add(1, std::sync::atomic::Ordering::SeqCst);

                if attempts < 2 {
                    Err(FError::RetryableError(Box::new(TestError("temporary error".to_string()))))
                } else {
                    Ok("success")
                }
            })
        });

        let retry_delay = std::time::Duration::from_millis(100);
        let retrying_fn = Composer::make_retry_callbox(
            failing_fn,
            3,
            retry_delay,
        );

        let start = Instant::now();
        let result = retrying_fn.run((), make_empty_ctx()).await;

        assert!(result.is_ok());
        assert_eq!(result.unwrap(), "success");
        assert_eq!(counter.load(std::sync::atomic::Ordering::SeqCst), 3);

        let elapsed = start.elapsed();
        assert!(elapsed >= retry_delay * 2, "执行时间应该至少包含两次重试延迟");

        // 添加明确的返回类型标注
        let non_retryable_fn: CallBox<'static, (), &'static str> = Composer::from_fn(|_: ()| {
            Box::pin(async {
                Err(FError::Generic(Box::new(std::io::Error::new(std::io::ErrorKind::InvalidData, "invalid format"))))
            })
        });

        let retrying_fn = Composer::make_retry_callbox(
            non_retryable_fn,
            3,
            retry_delay,
        );

        let result = retrying_fn.run((), make_empty_ctx()).await;
        assert!(matches!(result, Err(FError::Generic(_))));
    }

}
